package com.smsc.headend.task.engine.service.task.impl;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.SystemClock;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.toolkit.CollectionUtils;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.smsc.headend.common.annotation.TaskMessageCount;
import com.smsc.headend.common.constant.CacheNameConstant;
import com.smsc.headend.common.enums.DeviceType;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.asset.dmo.MeterConfigDMO;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.*;
import com.smsc.headend.module.task.entity.DeviceTask;
import com.smsc.headend.module.task.entity.DeviceTaskProcess;
import com.smsc.headend.module.task.entity.DeviceTaskSubProcess;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.manager.OdrRequestManager;
import com.smsc.headend.task.engine.mapper.DeviceTaskMapper;
import com.smsc.headend.task.engine.mapper.DeviceTaskProcessMapper;
import com.smsc.headend.task.engine.mapper.DeviceTaskSubProcessMapper;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import com.smsc.headend.task.engine.service.task.DeviceTaskService;
import com.smsc.headend.task.engine.service.task.DeviceTaskSubService;
import com.smsc.headend.task.engine.service.task.TaskDataService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.shardingsphere.driver.jdbc.core.datasource.ShardingSphereDataSource;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cache.annotation.*;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import javax.sql.DataSource;
import java.nio.charset.Charset;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Service
@Slf4j
@CacheConfig(cacheNames = CacheNameConstant.UTE_DEVICE_TASK)
@Transactional
public class DeviceTaskServiceImpl extends ServiceImpl<DeviceTaskMapper, DeviceTask> implements DeviceTaskService {
    public static final String UTF_8 = "utf-8";
    public static final String RESULT_NULL = "#result == null";
    @Autowired
    DeviceTaskSubService subService;
    @Autowired
    TaskDataService taskDataService;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    DataSource dataSource;
    @Autowired
    DeviceTaskProcessMapper deviceTaskProcessMapper;
    @Autowired
    DeviceTaskSubProcessMapper deviceTaskSubProcessMapper;
    @Autowired
    DeviceTaskService deviceTaskService;
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    OdrRequestManager odrRequestManager;
    @Autowired
    FeignAssetManagementService feignAssetManagementService;
    @Value("${spring.datasource.dbnames}")
    String[] dnNames;
    @Value("${spring.datasource.dbtype:oracle}")
    String dbType;


    @Override
    @Transactional
//    @CachePut(key = "#task.taskNo", unless = "#result == null")
//    @TaskMessageCount(type = "create")
    public DeviceTask addDeviceTask(Task task) {
//        if (lambdaQuery().eq(DeviceTask::getTaskNo, task.getTaskNo()).count() > 0) {
//            log.info("task {} already have record in log device task", task.getTaskNo());
//            return null;
//        }
        Long now = SystemClock.now() / 1000;
        DeviceTask deviceTask = new DeviceTask();
        deviceTask.setDeviceId(task.getMeterId());
//        deviceTask.setName(task.get);
        deviceTask.setTaskName(task.getTaskType());
        deviceTask.setTaskNo(task.getTaskNo());
        deviceTask.setPriority(task.getPriority());
        deviceTask.setTvCreate(now);
        deviceTask.setMessageId(task.getMessageId());
        deviceTask.setComId(task.getComId());
        deviceTask.setDeviceId(task.getMeterId());
        deviceTask.setStatus(TaskStatus.Waiting.toString());
        List<SubTaskGroup> subTaskGroups = task.getSubTaskGroups();
//        save(deviceTask);
        int seq = 0;
        taskDataService.setMessageIdTime(task.getMessageId());
        if (isTaskSubPersist(task)) {
            kafkaTemplate.send(TaskKafkaTopic.TASK_CREATE, JSONUtil.toJsonStr(deviceTask).getBytes(Charset.forName(UTF_8)));
            for (SubTaskGroup subTaskGroup : subTaskGroups) {
                subService.addTaskSub(task.getTaskNo(), subTaskGroup, now, seq++);
            }
        }
        return deviceTask;
    }

    @Override
    public void batchAddDeviceTask(List<DeviceTask> deviceTasks) {
//        this.saveBatch(deviceTasks);
        this.meterDeviceTasks(deviceTasks);
    }


    private Boolean meterDeviceTasksForOracle(List<DeviceTask> deviceTasks) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append("MERGE INTO LOG_DEVICE_TASK A")
                .append("    USING (SELECT ? AS TASK_NO, ? AS COM_ID, ? AS DEVICE_ID, ? AS TASK_NAME, ? AS TV_CREATE, ? AS MESSAGE_ID,? AS PRIORITY FROM DUAL) B ON ( A.TASK_NO = B.TASK_NO)")
                .append("    WHEN NOT MATCHED THEN INSERT(TASK_NO, COM_ID, DEVICE_ID, TASK_NAME, TV_CREATE, MESSAGE_ID, PRIORITY)")
                .append("    VALUES(B.TASK_NO, B.COM_ID, B.DEVICE_ID, B.TASK_NAME, B.TV_CREATE, B.MESSAGE_ID, B.PRIORITY)");

        String sql = stringBuilder.toString();


//        sql = String.format(sql, postfix);
        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (DeviceTask deviceTask : deviceTasks) {
                ps.setString(1, deviceTask.getTaskNo());
                ps.setLong(2, deviceTask.getComId());
                if (deviceTask.getDeviceId() == null) {
                    ps.setNull(3, Types.BIGINT);
                } else {
                    ps.setLong(3, deviceTask.getDeviceId());
                }
                ps.setString(4, deviceTask.getTaskName());
                ps.setLong(5, deviceTask.getTvCreate());
                ps.setString(6, deviceTask.getMessageId());
                ps.setInt(7, deviceTask.getPriority());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }

        return true;
    }

    private Boolean meterDeviceTasks(List<DeviceTask> deviceTasks) {
        if ("oracle".equals(dbType)) {
            return meterDeviceTasksForOracle(deviceTasks);
        } else if ("postgresql".equals(dbType)) {
            return meterDeviceTasksForPgsql(deviceTasks);
        } else {
            log.error("unknown db type");
        }
        return false;
    }

    private Boolean meterDeviceTasksForPgsql(List<DeviceTask> deviceTasks) {
        StringBuilder stringBuilder = new StringBuilder();
        stringBuilder.append(" with upsert as ");
        stringBuilder.append(" (     select A.TASK_NO from  LOG_DEVICE_TASK A , (SELECT ? AS TASK_NO, ? AS COM_ID, ? AS DEVICE_ID,? AS TASK_NAME, ? AS TV_CREATE, ? AS MESSAGE_ID,? AS PRIORITY ) B        where A.TASK_NO = B.TASK_NO) ");
        stringBuilder.append(" insert into LOG_DEVICE_TASK   select * from (SELECT ? AS TASK_NO, ? AS COM_ID, ? AS DEVICE_ID, ? AS TASK_NAME, ? AS TV_CREATE, ? AS MESSAGE_ID,? AS PRIORITY) t ");
        stringBuilder.append(" where not exists");
        stringBuilder.append(" (        select 1        from upsert B        where t.TASK_NO = B.TASK_NO )");
        String sql = stringBuilder.toString();

        DataSource defaultDbSource = ((ShardingSphereDataSource) dataSource).getContextManager().getDataSourceMap(((ShardingSphereDataSource) dataSource).getSchemaName()).get(dnNames[0]);

//        sql = String.format(sql, postfix);

        try (Connection conn = defaultDbSource.getConnection(); PreparedStatement ps = conn.prepareStatement(sql)) {
            conn.setAutoCommit(false);
            for (DeviceTask deviceTask : deviceTasks) {
                ps.setString(1, deviceTask.getTaskNo());
                ps.setLong(2, deviceTask.getComId());
                if (deviceTask.getDeviceId() == null) {
                    ps.setNull(3, Types.BIGINT);
                } else {
                    ps.setLong(3, deviceTask.getDeviceId());
                }
                ps.setString(4, deviceTask.getTaskName());
                ps.setLong(5, deviceTask.getTvCreate());
                ps.setString(6, deviceTask.getMessageId());
                ps.setInt(7, deviceTask.getPriority());
                ps.setString(8, deviceTask.getTaskNo());
                ps.setLong(9, deviceTask.getComId());
                if (deviceTask.getDeviceId() == null) {
                    ps.setNull(10, Types.BIGINT);
                } else {
                    ps.setLong(10, deviceTask.getDeviceId());
                }
                ps.setString(11, deviceTask.getTaskName());
                ps.setLong(12, deviceTask.getTvCreate());
                ps.setString(13, deviceTask.getMessageId());
                ps.setInt(14, deviceTask.getPriority());
                ps.addBatch();
            }
            ps.executeBatch();
            conn.commit();
            ps.close();
            conn.close();
        } catch (SQLException throwables) {
            log.error("batch insert error", throwables);
            return false;
        }

        return true;
    }

    @Override
    public boolean isTaskSubPersist(Task task) {
        if (TaskType.ScheduleRead_Profile.toString().equalsIgnoreCase(task.getTaskType())) {
            return false;
        }

        if (TaskType.EventRead_Profile.toString().equalsIgnoreCase(task.getTaskType()) && !task.getIsODR()) {
            return false;
        }

        if (TaskType.System_Time_Sync.toString().equalsIgnoreCase(task.getTaskType()) && !task.getIsODR()) {
            return false;
        }

        if (TaskType.AutoReCollectProfile.toString().equalsIgnoreCase(task.getTaskType())) {
            return false;
        }

        if (TaskType.Schedule_Read.toString().equalsIgnoreCase(task.getTaskType()) ||
                TaskType.ScheduleGet_Capture_OBJS.toString().equalsIgnoreCase(task.getTaskType())) {
            return false;
        }

        if (TaskType.GetOnlineMeterList.toString().equalsIgnoreCase(task.getTaskType())) {
            return false;
        }

        return true;
    }

    @Override
    public void updateSubDeviceTaskStart(Task task, AtomicTask atomicTask) {
//        DeviceTask deviceTask = lambdaQuery().eq(DeviceTask::getTaskNo, task.getTaskNo()).one();
        Integer subTaskSeq = 0;
        for (SubTaskGroup subTaskGroup : task.getSubTaskGroups()) {
            //start SubTaskGroup
            if (subTaskGroup.getSubTasks().get(0).getAtomicTaskNo().equalsIgnoreCase(atomicTask.getAtomicTaskNo())) {
                //start Task
                if (isTaskSubPersist(task)) {
                    sendSubTaskMessageUpdate(task, subTaskSeq, SubTaskStatus.Processing.getCode());
                }
                break;
            }
            subTaskSeq++;
        }
    }

    public void sendSubTaskMessageUpdate(String taskNo, Integer sort, int code) {
        SubTaskStatusDto subDto = new SubTaskStatusDto();
        subDto.setTaskNo(taskNo);
        subDto.setStatus(code);
        subDto.setSort(sort);
        subDto.setTimestamp(SystemClock.now() / 1000);
        kafkaTemplate.send(TaskKafkaTopic.SUB_TASK_DEVICE_STATUS_UPDATE, JSONUtil.toJsonStr(subDto).getBytes(Charset.forName(UTF_8)));
    }

    public void sendSubTaskMessageUpdate(Task task, Integer sort, int code) {
        sendSubTaskMessageUpdate(task.getTaskNo(), sort, code);
    }

    @Override
//    @TaskMessageCount(type = "update")
    // @CachePut(key = "'status:'  + #task.taskNo", unless = "#result == null")
    public Integer sendTaskStatusUpdateMessage(Task task, String messageId, TaskStatus status) {
        redisUtils.set("task.device::status:" + task.getTaskNo(), status.getCode(), 3600);
        if (!isTaskSubPersist(task)) {
            return status.getCode();
        }
        TaskStatusDto dto = new TaskStatusDto();
        dto.setTaskNo(task.getTaskNo());
        dto.setTaskStatus(status.getCode());
        dto.setMessageId(messageId);
        dto.setTimestamp(SystemClock.now() / 1000);
        kafkaTemplate.send(TaskKafkaTopic.TASK_DEVICE_STATUS_UPDATE, JSONUtil.toJsonStr(dto).getBytes(Charset.forName(UTF_8)));
        deviceTaskService.updateTaskTvAndStatusCache(dto);
        return status.getCode();
    }

    /**
     * @param taskNo
     * @return
     */
    @Override
//    @Cacheable(key = "#task.taskNo", unless = "#result == null")
    public DeviceTask getTaskByTaskNo(String taskNo) {
        return lambdaQuery().eq(DeviceTask::getTaskNo, taskNo).one();
    }


    @Override
    public void updateRequestDeviceTask(Task task, Integer subTaskStatus) {
        if (StringUtils.isEmpty(task.getMessageId())) {
            log.error("update message id null");
            return;
        }
        if (isTaskSubPersist(task)) {
            sendTaskStatusUpdateMessage(task, task.getMessageId(), TaskStatus.Complete);
            List<DeviceTask> tasks = odrRequestManager.getDeviceTask(task.getMessageId());
            log.info("task execute: failed messageId={}", task.getMessageId());
            if (tasks == null) {
                return;
            }
            tasks.stream().forEach(subTask -> updateDeviceTaskCancel(subTask, subTaskStatus));
            tasks = null;
        }
    }

    public void updateDeviceTaskCancel(DeviceTask task, Integer subTaskStatus) {
        sendSubTaskMessageUpdate(task.getTaskNo(), null, subTaskStatus);
    }

    @Override
    @Cacheable(key = "'status:'  + #taskNo", unless = RESULT_NULL)
    public Integer getTaskStatus(String taskNo) {
        List<DeviceTaskProcess> deviceTaskProcess = deviceTaskProcessMapper.selectList(Wrappers.<DeviceTaskProcess>lambdaQuery().eq(DeviceTaskProcess::getTaskNo, taskNo));
        if (CollectionUtils.isEmpty(deviceTaskProcess)) {
            return TaskStatus.Waiting.getCode();
        }
        deviceTaskProcess.sort((o1, o2) -> {
            int compareResult = o2.getTv().compareTo(o1.getTv());
            if (compareResult == 0) {
                return o2.getStatus().compareTo(o1.getStatus());
            }
            return compareResult;
        });
        return deviceTaskProcess.get(0).getStatus();
    }

    @Override
    @CachePut(key = "'tv.status:'  + #dto.taskNo", unless = RESULT_NULL)
    public DeviceTask updateTaskTvAndStatusCache(TaskStatusDto dto) {
        DeviceTask result = new DeviceTask();
        if (dto.getTaskStatus() == TaskStatus.Processing.getCode()) {
            result.setTaskNo(dto.getTaskNo());
            result.setStatus(TaskStatus.Processing.toString());
            result.setTvStart(dto.getTimestamp());
            return result;
        }
        DeviceTask deviceTask = deviceTaskService.getTaskTvAndStatus(dto.getTaskNo());
        if (dto.getTaskStatus() == TaskStatus.Complete.getCode()) {
            deviceTask.setTvEnd(dto.getTimestamp());
        }
        deviceTask.setStatus(TaskStatus.forValue(dto.getTaskStatus()).toString());
        return deviceTask;
    }

    @Override
    @Cacheable(key = "'est.sec' + #comId + ':' + #deviceId", unless = RESULT_NULL)
    public String estimateSingleDeviceTaskTime(Long comId, Long deviceId) {
        List<DeviceTaskSubProcess> recentTaskExecuteInfo = new ArrayList<>();
        if (dbType.equals("oracle")) {
            recentTaskExecuteInfo = deviceTaskSubProcessMapper.getRecentTaskExecuteInfo(comId, deviceId);
        } else if (dbType.equals("postgresql")) {
            recentTaskExecuteInfo = deviceTaskSubProcessMapper.getRecentTaskExecuteInfoForPgsql(comId, deviceId);
        }
        log.info("SingleTaskTimeEstimate,comId:{},deviceId:{},subTaskSize:{}", comId, deviceId, recentTaskExecuteInfo.size());
        return estimateTime(recentTaskExecuteInfo);
    }

    @Override
    @CacheEvict(key = "'est.sec' + #comId + ':' + #deviceId")
    public void clearEstimateCache(Long comId, Long deviceId) {
        log.info("clearEstimateCache:comId{},#{}", comId, deviceId);
    }

    private String estimateTime(List<DeviceTaskSubProcess> recentTaskExecuteInfo) {
        Map<String, List<DeviceTaskSubProcess>> collect = recentTaskExecuteInfo.stream().collect(Collectors.groupingBy(DeviceTaskSubProcess::getTaskNo));
        List<Double> costTimeList = collect.entrySet().stream().map(
                kv -> {
                    List<DeviceTaskSubProcess> value = kv.getValue();
                    Double executeSeconds = subTaskCostTime(value);
                    return executeSeconds;
                }
        ).filter(s -> s != null).collect(Collectors.toList());
        double v = costTimeList.stream().mapToDouble(Double::doubleValue).average().orElse(-1);
        return Convert.toStr(v);
    }

    private Double subTaskCostTime(List<DeviceTaskSubProcess> value) {
        int maxSort = value.stream().filter(d -> d.getSort() != null)
                .filter(d -> d.getStatus() == SubTaskStatus.Waiting.getCode() || d.getStatus() == SubTaskStatus.Success.getCode())
                .map(DeviceTaskSubProcess::getSort).mapToInt(Integer::intValue).max().orElse(-1);
        if (maxSort == -1) {
            return null;
        }
        List<Long> durations = new ArrayList<>();
        for (int j = 0; j <= maxSort; j++) {
            int finalJ = j;
            List<Long> tvs = value.stream().filter(d -> d.getSort() != null)
                    .filter(dtsp -> dtsp.getSort().intValue() == finalJ).map(DeviceTaskSubProcess::getTv).collect(Collectors.toList());
            if (tvs == null || tvs.size() < 2) {
                continue;
            }
            long end = tvs.stream().mapToLong(Long::longValue).max().getAsLong();
            long start = tvs.stream().mapToLong(Long::longValue).min().getAsLong();
            long duration = end - start;
            durations.add(duration == 0 ? 1 : duration);
        }
        double avg = durations.stream().mapToLong(Long::longValue).average().orElse(-1);
        return avg < 0 ? null : avg;
    }

    @Override
    @Cacheable(key = "'tv.status:'  + #taskNo", unless = RESULT_NULL)
    public DeviceTask getTaskTvAndStatus(String taskNo) {
        List<DeviceTaskProcess> deviceTaskProcess = deviceTaskProcessMapper.selectList(Wrappers.<DeviceTaskProcess>lambdaQuery().eq(DeviceTaskProcess::getTaskNo, taskNo));
        DeviceTask result = new DeviceTask();
        if (CollectionUtils.isEmpty(deviceTaskProcess)) {
            result.setStatus(TaskStatus.Waiting.toString());
            return result;
        }

        deviceTaskProcess.sort((o1, o2) -> {
            int compareResult = o2.getTv().compareTo(o1.getTv());
            if (compareResult == 0) {
                return o2.getStatus().compareTo(o1.getStatus());
            }
            return compareResult;
        });

        result.setTaskNo(taskNo);
        Integer status = deviceTaskProcess.get(0).getStatus();
        result.setStatus(TaskStatus.forValue(status).toString());
        DeviceTaskProcess startingProcess = deviceTaskProcess.stream().filter(o -> o.getStatus() == TaskStatus.Processing.getCode()).findFirst().orElse(null);
        DeviceTaskProcess endProcess = deviceTaskProcess.stream().filter(o -> o.getStatus() == TaskStatus.Complete.getCode()).findFirst().orElse(null);
        result.setTvStart(startingProcess.getTv());
        result.setTvEnd(endProcess == null ? null : endProcess.getTv());
        return result;
    }

    @Override
    @Transactional
    public void batchInsertTaskProcess(List<TaskStatusDto> dtoList) {
        List<DeviceTaskProcess> deviceTaskProcesses = dtoList.stream().map(d -> {
            DeviceTaskProcess p = new DeviceTaskProcess();
            p.setStatus(d.getTaskStatus());
            p.setTv(d.getTimestamp());
            p.setTaskNo(d.getTaskNo());
            return p;
        }).collect(Collectors.toList());

        this.executeBatch(deviceTaskProcesses, (sqlSession, dtp) -> {
            sqlSession.insert("com.smsc.headend.task.engine.mapper.DeviceTaskProcessMapper.insert", dtp);
        });
    }


    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void batchInsertSubTaskStatus(List<SubTaskStatusDto> dtoList) {

        List<DeviceTaskSubProcess> deviceTaskSubProcesses = dtoList.stream().map(d -> {
            DeviceTaskSubProcess p = new DeviceTaskSubProcess();
            p.setStatus(d.getStatus());
            p.setTv(d.getTimestamp());
            p.setTaskNo(d.getTaskNo());
            p.setSort(d.getSort());
            return p;
        }).collect(Collectors.toList());
        this.executeBatch(deviceTaskSubProcesses, (session, dto) -> {
            session.insert("com.smsc.headend.task.engine.mapper.DeviceTaskSubProcessMapper.insert", dto);
        });
    }
}
